package org.zhaya.sdk.util;

import cn.hutool.json.JSONUtil;
import com.fasterxml.jackson.core.json.JsonReadFeature;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.zhaya.sdk.swamill.CustomPipeline;
import org.zhaya.sdk.swamill.FactoriesRegister;
import org.zhaya.sdk.swamill.ProcessDoc;

import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @author jiayongming
 */
@Slf4j
public class PipelineUtils {
    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper()
            .configure(JsonReadFeature.ALLOW_UNESCAPED_CONTROL_CHARS.mappedFeature(), true)
            .configure(JsonReadFeature.ALLOW_LEADING_ZEROS_FOR_NUMBERS.mappedFeature(), true)
            .configure(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS, true);
    private static final TypeReference<List<Map<String, Object>>> LIST_TYPE_REFERENCE = new TypeReference<List<Map<String, Object>>>() {
    };
    private static final TypeReference<Map<String, Object>> MAP_TYPE_REFERENCE = new TypeReference<Map<String, Object>>() {
    };
    private static final String STOP_ON_FAILURE_TRUE = "\"stopOnFailure\":true";
    private static final String STOP_ON_FAILURE_FALSE = "\"stopOnFailure\":false";

    public static List<ProcessDoc> createDocsForMessage(String data) {
        if (JSONUtil.isTypeJSONArray(data)) {
            try {
                List<Map<String, Object>> list = OBJECT_MAPPER.readValue(data, LIST_TYPE_REFERENCE);
                if (list.isEmpty()) {
                    return Collections.emptyList();
                }
                return list.stream().map(ProcessDoc::new).collect(Collectors.toList());
            } catch (Exception e) {
                log.trace("PipelineUtils cannot parse array data {} to json.", data);
                return Lists.newArrayList(createDocForMessage(data));
            }
        } else {
            return Lists.newArrayList(createDocForMessage(data));
        }
    }

    public static ProcessDoc createDocForMessage(String data) {
        if (!JSONUtil.isTypeJSONObject(data)) {
            throw new RuntimeException("raw data can not parse to json");
        }
        try {
            log.trace("data {} will be processed as json.", data);
            return new ProcessDoc(OBJECT_MAPPER.readValue(data, MAP_TYPE_REFERENCE));
        } catch (Exception e) {
            log.warn("PipelineUtils cannot parse data {} to json since error {}", data, e.getMessage());
            throw new RuntimeException("raw data can not parse to json");
        }
    }

    public static CustomPipeline getPipeline(String id, String conf) {
        return new CustomPipeline.Factory(FactoriesRegister.processorFactoryRegistry, FactoriesRegister.conditionFactoryRegistry).create(id, conf);
    }

    /**
     * 默认配置stopOnFailure=true
     *
     * @param id
     * @param conf
     * @param stopOnFailure
     * @return
     */
    public static CustomPipeline getPipeline(String id, String name, String conf, boolean stopOnFailure) {
        String stepConfig = conf;
        if (!stopOnFailure) {
            stepConfig = conf.replace(STOP_ON_FAILURE_TRUE, STOP_ON_FAILURE_FALSE);
        }
        return new CustomPipeline.Factory(FactoriesRegister.processorFactoryRegistry, FactoriesRegister.conditionFactoryRegistry).create(id, name, stepConfig);
    }
}